package com.yl.util;

import cn.hutool.core.date.DateUtil;
import com.yl.constant.CdcType;
import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import com.yl.entity.Packet;
import com.yl.entity.cdc.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;

import java.util.*;

@Slf4j
public class PayloadParseUtil {

    /**
     * 解析原始数据，封装成 List<MultiDataEntity> 对象
     *
     * @param payload     流元素
     * @param configState 配置信息
     * @return List<MultiDataEntity>
     */
    public static List<MultiDataEntity> parse(String payload, ReadOnlyBroadcastState<String, List<MysqlCdcEntity>> configState) throws Exception {
        List<MultiDataEntity> res = new ArrayList<>();
        // 获取功能码
        int funCode = Integer.parseInt(payload.split(Const.PTN_COMMA)[2]);
        if (funCode == 1) {
            // 功能码1
            res.addAll(parseF1(payload, configState));
        } else if (funCode == 2) {
            // 功能码2
            res.addAll(parseF2(payload, configState));
        } else if (funCode == 3) {
            // 功能码3
            res.addAll(parseF3(payload, configState));
        }
        return res;
    }

    /**
     * 解析原始数据【功能码1】
     * 原始流元素{项目id,时间（17位字符串）,功能码(1),测点类型,预留字节,测点数,指标数,n组数据}
     * 注意：n组数据中每组m个数，第一个为测点id，后m-1个为该测点指标数据，其中测点数=n，指标数=m-1
     *
     * @param payload 流元素
     * @return List<MultiDataEntity>
     */
    private static List<MultiDataEntity> parseF1(String payload, ReadOnlyBroadcastState<String, List<MysqlCdcEntity>> configState) throws Exception {
        List<MultiDataEntity> res = new ArrayList<>();
        String errMsg;
        // 测点数，测点类型，指标数
        int sensorNum, typeId, quotaNum;
        String[] items = payload.split(Const.PTN_COMMA);
        int elementSize = items.length;
        // 判断基本信息
        if (elementSize < 9) {
            errMsg = "原始数据格式有误：信息内容长度小于9【信息内容长度：{}】";
            log.error(errMsg, elementSize);
            return res;
        }
        // 解析数据基本信息
        int index = 0;
        String projectId = items[index++]; // 项目id
        String key = SUtil.fmt(CdcType.PRO.getTable(), projectId);
        BaseProject baseProject = ClassUtil.cast(configState.get(key).get(0), BaseProject.class);
        // 判断项目信息
        if (null == baseProject) {
            errMsg = "原始数据格式有误：项目id不存在【项目id：{}】";
            log.error(errMsg, projectId);
            return res;
        }
        // 根据项目id获取用户信息
        key = SUtil.fmt(CdcType.USR.getTable(), projectId);
        List<AuthUser> users = ClassUtil.cast(configState.get(key), AuthUser.class);
        // 判断数据时间
        String cjTime = items[index++]; // 数据采集时间
        if (cjTime.length() != 17) {
            errMsg = "原始数据格式有误：数据时间长度不是17位【数据时间：{}，长度：{}】";
            log.error(errMsg, cjTime, cjTime.length());
            return res;
        }
        long pickTime = DateUtil.parse(cjTime, Const.FMT_TRIM_MILLI).getTime() * 1000;
        index++; // 功能码
        typeId = Integer.parseInt(items[index++]); // 测点类型id
        key = SUtil.fmt(CdcType.TYP.getTable(), typeId);
        BaseTargetType baseTargetType = ClassUtil.cast(configState.get(key).get(0), BaseTargetType.class);
        // 判断数据类型
        if (null == baseTargetType) {
            errMsg = "原始数据格式有误：测点类型不存在【测点类型id：{}】";
            log.error(errMsg, typeId);
            return res;
        }
        index++; // 预留字节
        sensorNum = Integer.parseInt(items[index++]); // 测点个数
        quotaNum = Integer.parseInt(items[index]); // 指标个数
        // 判断测点及指标个数
        int valueSize = sensorNum * (quotaNum + 1);
        if (elementSize != 7 + valueSize) {
            errMsg = "原始数据格式有误：测点和指标标注个数与统计个数不符【标注个数：{}，统计个数：{}】";
            log.error(errMsg, valueSize, elementSize - 7);
            return res;
        }
        // 获取指标的单位信息和计算信息
        List<Integer> qcs = new ArrayList<>();
        List<SettingsDataQuota> settingsDataQuotas = new ArrayList<>();
        for (int i = 0; i < quotaNum; i++) {
            key = SUtil.fmt(CdcType.QTA.getTable(), SUtil.fmt(typeId, i + 1));
            Optional
                    .ofNullable(configState.get(key))
                    .flatMap(qts -> Optional.ofNullable(ClassUtil.cast(qts.get(0), SettingsDataQuota.class)))
                    .ifPresent(settingsDataQuota -> {
                        settingsDataQuotas.add(settingsDataQuota);
                        if (null == settingsDataQuota.getCal_type_code()) {
                            qcs.add(settingsDataQuota.getCode());
                        }
                    });
        }
        try {
            // 解析指标数据
            for (int i = 7; i <= elementSize - quotaNum; i += (quotaNum + 1)) {
                String serialNumber = items[i++];
                // 获取测点信息
                key = SUtil.fmt(CdcType.TGT.getTable(), SUtil.fmt(projectId, serialNumber));
                BaseTarget baseTarget = ClassUtil.cast(configState.get(key).get(0), BaseTarget.class);
                if (null == baseTarget) {
                    errMsg = "原始数据格式有误：测点通讯编号不存在【项目id：{}，通讯编号：{}】";
                    log.error(errMsg, projectId, serialNumber);
                    continue;
                }
                // 存储的指标数据
                Map<Integer, Double> rawValues = new HashMap<>();
                Map<Integer, List<SettingsThreshold>> ths = new HashMap<>();
                for (int qc = 1; qc <= quotaNum; qc++) {
                    // 只存储 enable=1 的指标
                    if (qcs.contains(qc)) {
                        rawValues.put(qc, Double.parseDouble(items[i + qc - 1]));
                        // 获取该指标的阈值信息
                        key = SUtil.fmt(CdcType.THD.getTable(), SUtil.fmt(baseTarget.getId(), qc));
                        List<MysqlCdcEntity> thEntity = configState.get(key);
                        if (null != thEntity) {
                            ths.put(qc, ClassUtil.cast(thEntity, SettingsThreshold.class));
                        }
                    }
                }
                MultiDataEntity multiDataEntity = MultiDataEntity.builder()
                        .project(baseProject)
                        .users(users)
                        .target(baseTarget)
                        .targetType(baseTargetType)
                        .quotas(settingsDataQuotas)
                        .pickTime(pickTime)
                        .rawValues(rawValues)
                        .ths(ths)
                        .build();
                res.add(multiDataEntity);
            }
        } catch (Exception e) {
            errMsg = "原始数据格式有误：测点及指标信息解析失败";
            log.error(errMsg);
            e.printStackTrace();
        }
        return res;
    }

    /**
     * 解析原始数据【功能码2】
     * 项目id,时间(17位字符串),功能码(2),测点类型,采集周期p(ms),数据组数n(n=1000/p),指标数m,测点通讯id,n组数据(每组m个)
     * 1,20220124173459022,2,2000,50,4,1,2,0.11,0.22,0.33,0.44
     *
     * @param payload 流元素
     * @return List<MultiDataEntity>
     */
    private static List<MultiDataEntity> parseF2(String payload, ReadOnlyBroadcastState<String, List<MysqlCdcEntity>> configState) throws Exception {
        List<MultiDataEntity> res = new ArrayList<>();
        String errMsg;
        // 测点类型，指标个数
        int typeId, quotaNum;
        String[] items = payload.split(Const.PTN_COMMA);
        int elementSize = items.length;
        // 判断基本信息
        if (elementSize < 9) {
            errMsg = "原始数据格式有误：信息内容长度小于9【信息内容长度：{}】";
            log.error(errMsg, elementSize);
            return res;
        }
        // 解析数据基本信息
        int index = 0;
        String projectId = items[index++]; // 项目id
        String key = SUtil.fmt(CdcType.PRO.getTable(), projectId);
        BaseProject baseProject = ClassUtil.cast(configState.get(key).get(0), BaseProject.class);
        // 判断项目信息
        if (null == baseProject) {
            errMsg = "原始数据格式有误：项目id不存在【项目id：{}】";
            log.error(errMsg, projectId);
            return res;
        }
        // 根据项目id获取用户信息
        key = SUtil.fmt(CdcType.USR.getTable(), projectId);
        List<AuthUser> users = ClassUtil.cast(configState.get(key), AuthUser.class);
        // 判断数据时间
        String cjTime = items[index++]; // 数据采集时间
        if (cjTime.length() != 17) {
            errMsg = "原始数据格式有误：数据时间长度不是17位【数据时间：{}，长度：{}】";
            log.error(errMsg, cjTime, cjTime.length());
            return res;
        }
        // 数据采集时间起始时间
        long pickStartTime = DateUtil.parse(cjTime, Const.FMT_TRIM_MILLI).getTime() * 1000;
        index++; // 功能码
        typeId = Integer.parseInt(items[index++]); // 测点类型id
        key = SUtil.fmt(CdcType.TYP.getTable(), typeId);
        BaseTargetType baseTargetType = ClassUtil.cast(configState.get(key).get(0), BaseTargetType.class);
        // 判断数据类型
        if (null == baseTargetType) {
            errMsg = "原始数据格式有误：测点类型不存在【测点类型id：{}】";
            log.error(errMsg, typeId);
            return res;
        }
        double cjp = Double.parseDouble(items[index++]); // 采集周期
        int groups = Integer.parseInt(items[index++]); // 数据组数
        quotaNum = Integer.parseInt(items[index++]); // 指标数
        // 判断测点及指标个数
        int valueSize = quotaNum * groups;
        if (elementSize != 8 + valueSize) {
            errMsg = "原始数据格式有误：测点和指标标注个数与统计个数不符【标注个数：{}，统计个数：{}】";
            log.error(errMsg, valueSize, elementSize - 8);
            return res;
        }
        // 解析测点指标数据
        String serialNumber = items[index]; // 测点通讯id
        // 获取测点信息
        key = SUtil.fmt(CdcType.TGT.getTable(), SUtil.fmt(projectId, serialNumber));
        BaseTarget baseTarget = ClassUtil.cast(configState.get(key).get(0), BaseTarget.class);
        if (null == baseTarget) {
            errMsg = "原始数据格式有误：测点通讯编号不存在【项目id：{}，通讯编号：{}】";
            log.error(errMsg, projectId, serialNumber);
            return res;
        }
        List<Integer> qcs = new ArrayList<>();
        Map<Integer, List<SettingsThreshold>> ths = new HashMap<>();
        List<SettingsDataQuota> settingsDataQuotas = new ArrayList<>();
        for (int qc = 1; qc <= quotaNum; qc++) {
            // 获取指标的单位信息和计算信息
            key = SUtil.fmt(CdcType.QTA.getTable(), SUtil.fmt(typeId, qc));
            Optional
                    .ofNullable(configState.get(key))
                    .flatMap(qts -> Optional.ofNullable(ClassUtil.cast(qts.get(0), SettingsDataQuota.class)))
                    .ifPresent(settingsDataQuota -> {
                        settingsDataQuotas.add(settingsDataQuota);
                        if (null == settingsDataQuota.getCal_type_code()) {
                            qcs.add(settingsDataQuota.getCode());
                        }
                    });
            // 获取该指标的阈值信息
            key = SUtil.fmt(CdcType.THD.getTable(), SUtil.fmt(baseTarget.getId(), qc));
            List<MysqlCdcEntity> thEntity = configState.get(key);
            if (null != thEntity) {
                ths.put(qc, ClassUtil.cast(thEntity, SettingsThreshold.class));
            }
        }
        try {
            // 解析指标数据
            for (int i = 8; i <= elementSize - quotaNum; i += quotaNum) {
                // 存储的指标数据
                Map<Integer, Double> rawValues = new HashMap<>();
                for (int qc = 1; qc <= quotaNum; qc++) {
                    // 只存储 enable=1 的指标
                    if (qcs.contains(qc)) {
                        int qvi = i + qc - 1;
                        double qv = Double.parseDouble(items[qvi]);
                        rawValues.put(qc, qv);
                    }
                }
                long pickTime = getDataTime(pickStartTime, cjp, (i - 8) / quotaNum);
                MultiDataEntity multiDataEntity = MultiDataEntity.builder()
                        .project(baseProject)
                        .users(users)
                        .target(baseTarget)
                        .targetType(baseTargetType)
                        .quotas(settingsDataQuotas)
                        .pickTime(pickTime)
                        .rawValues(rawValues)
                        .ths(ths)
                        .build();
                res.add(multiDataEntity);
            }
        } catch (Exception e) {
            errMsg = "原始数据格式有误：测点及指标信息解析失败";
            log.error(errMsg);
            e.printStackTrace();
        }
        return res;
    }

    /**
     * 解析原始数据【功能码3】
     * 项目id,时间(20位字符串),功能码(3),测点类型,采集周期p(ms),包批次,包总数(与采集时长和发送频率有关),包序号,
     * 指标数m,测点通讯id,n组数据(n是m的整数倍，但具体数值与发送频率有关)
     * 1,20220124173459084011,3,2000,50,1,100,1,1,2,0.11,0.22,0.33,0.44,0.55,0.66
     *
     * @param payload 流元素
     * @return List<MultiDataEntity>
     */
    private static List<MultiDataEntity> parseF3(String payload, ReadOnlyBroadcastState<String, List<MysqlCdcEntity>> configState) throws Exception {
        List<MultiDataEntity> res = new ArrayList<>();
        String errMsg;
        // 包批次，包总数，包序号，测点类型，指标个数
        int batchId, packetSum, packetId, typeId, quotaNum;
        String[] items = payload.split(Const.PTN_COMMA);
        int elementSize = items.length;
        // 判断基本信息
        if (elementSize < 11) {
            errMsg = "原始数据格式有误：信息内容长度小于11【信息内容长度：{}】";
            log.error(errMsg, elementSize);
            return res;
        }
        // 解析数据基本信息
        int index = 0;
        String projectId = items[index++]; // 项目id
        String key = SUtil.fmt(CdcType.PRO.getTable(), projectId);
        BaseProject baseProject = ClassUtil.cast(configState.get(key).get(0), BaseProject.class);
        // 判断项目信息
        if (null == baseProject) {
            errMsg = "原始数据格式有误：项目id不存在【项目id：{}】";
            log.error(errMsg, projectId);
            return res;
        }
        // 根据项目id获取用户信息
        key = SUtil.fmt(CdcType.USR.getTable(), projectId);
        List<AuthUser> users = ClassUtil.cast(configState.get(key), AuthUser.class);
        // 判断数据时间
        String cjTime = items[index++]; // 数据采集时间
        if (cjTime.length() != 20) {
            errMsg = "原始数据格式有误：数据时间长度不是20位【数据时间：{}，长度：{}】";
            log.error(errMsg, cjTime, cjTime.length());
            return res;
        }
        // 数据采集时间起始时间
        long pickStartTime = DtUtil.parseDateStr(cjTime);
        index++; // 功能码
        typeId = Integer.parseInt(items[index++]); // 测点类型id
        key = SUtil.fmt(CdcType.TYP.getTable(), typeId);
        BaseTargetType baseTargetType = ClassUtil.cast(configState.get(key).get(0), BaseTargetType.class);
        // 判断数据类型
        if (null == baseTargetType) {
            errMsg = "原始数据格式有误：测点类型不存在【测点类型id：{}】";
            log.error(errMsg, typeId);
            return res;
        }
        double cjp = Double.parseDouble(items[index++]); // 采集周期
        batchId = Integer.parseInt(items[index++]); // 包批次
        packetSum = Integer.parseInt(items[index++]); // 包总数
        packetId = Integer.parseInt(items[index++]); // 包序号
        // 判断包总数和序号
        if (packetId > packetSum) {
            errMsg = "原始数据格式有误：数据包序号应不大于包总数【包总数：{}，包序号：{}】";
            log.error(errMsg, packetSum, packetId);
            return res;
        }
        quotaNum = Integer.parseInt(items[index++]); // 指标数
        // 判断指标个数与数据组数的关系
        int valueSize = elementSize - 10;
        if (valueSize % quotaNum != 0) {
            errMsg = "原始数据格式有误：数据总数应是指标个数的整数倍【指标个数：{}，数据总数：{}】";
            log.error(errMsg, quotaNum, valueSize);
            return res;
        }
        // 解析测点指标数据
        String serialNumber = items[index]; // 测点通讯id
        // 获取测点信息
        key = SUtil.fmt(CdcType.TGT.getTable(), SUtil.fmt(projectId, serialNumber));
        BaseTarget baseTarget = ClassUtil.cast(configState.get(key).get(0), BaseTarget.class);
        if (null == baseTarget) {
            errMsg = "原始数据格式有误：测点通讯编号不存在【项目id：{}，通讯编号：{}】";
            log.error(errMsg, projectId, serialNumber);
            return res;
        }
        // 获取指标的单位信息和计算信息
        List<Integer> qcs = new ArrayList<>();
        Map<Integer, List<SettingsThreshold>> ths = new HashMap<>();
        List<SettingsDataQuota> settingsDataQuotas = new ArrayList<>();
        for (int qc = 1; qc <= quotaNum; qc++) {
            key = SUtil.fmt(CdcType.QTA.getTable(), SUtil.fmt(typeId, qc));
            Optional
                    .ofNullable(configState.get(key))
                    .flatMap(qts -> Optional.ofNullable(ClassUtil.cast(qts.get(0), SettingsDataQuota.class)))
                    .ifPresent(settingsDataQuota -> {
                        settingsDataQuotas.add(settingsDataQuota);
                        if (null == settingsDataQuota.getCal_type_code()) {
                            qcs.add(settingsDataQuota.getCode());
                        }
                    });
            // 获取该指标的阈值信息
            key = SUtil.fmt(CdcType.THD.getTable(), SUtil.fmt(baseTarget.getId(), qc));
            List<MysqlCdcEntity> thEntity = configState.get(key);
            if (null != thEntity) {
                ths.put(qc, ClassUtil.cast(thEntity, SettingsThreshold.class));
            }
        }
        try {
            boolean noPacketFlag = true;
            long packetStartTime = getDataTime(pickStartTime, cjp, 0);
            long packetEndTime = getDataTime(pickStartTime, cjp, (valueSize - quotaNum) / quotaNum);
            // 解析指标数据
            for (int i = 10; i <= elementSize - quotaNum; i += quotaNum) {
                // 数据包信息
                Packet packet = null;
                long pickTime = getDataTime(pickStartTime, cjp, (i - 10) / quotaNum);
                // 每个数据包输出一个带包信息的流元素，其他包信息为null
                if (noPacketFlag) {
                    // 该数据包信息
                    packet = Packet.builder()
                            .packetId(packetId)
                            .batchId(batchId)
                            .packetSum(packetSum)
                            .packetStartTime(packetStartTime / 1000)
                            .packetEndTime(packetEndTime / 1000)
                            .build();
                    noPacketFlag = false;
                }
                // 存储的指标数据
                Map<Integer, Double> rawValues = new HashMap<>();
                for (int qc = 1; qc <= quotaNum; qc++) {
                    // 只存储 enable=1 的指标
                    if (qcs.contains(qc)) {
                        int qvi = i + qc - 1;
                        double qv = Double.parseDouble(items[qvi]);
                        rawValues.put(qc, qv);
                    }
                }
                MultiDataEntity multiDataEntity = MultiDataEntity.builder()
                        .project(baseProject)
                        .users(users)
                        .target(baseTarget)
                        .targetType(baseTargetType)
                        .ths(ths)
                        .quotas(settingsDataQuotas)
                        .pickTime(pickTime)
                        .rawValues(rawValues)
                        .packet(packet)
                        .build();
                res.add(multiDataEntity);
            }
        } catch (Exception e) {
            errMsg = "原始数据格式有误：测点及指标信息解析失败";
            log.error(errMsg);
            e.printStackTrace();
        }
        return res;
    }

    /**
     * @param pickTime 数据开始时间
     * @param cjp      采集周期
     * @return 返回偏移后的数据时间
     */
    private static long getDataTime(long pickTime, double cjp, int offset) {
        return pickTime + (long) (cjp * 1000 * offset);
    }

}
